package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class LinearRegressionReducer extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {

	static long DAY_IN_MS = 24 * 60 * 60 * 1000;

	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {

		TimestampPriceDataPoint nextPoint;

		int windowSizeInDays = context.getConfiguration().getInt("windowSize", CONFIG.WINDOW_SIZE);
		int windowStepSizeInDays = context.getConfiguration().getInt("windowStepSize", 1);

		long windowSizeInMS = windowSizeInDays * DAY_IN_MS;
		long windowStepSizeInMS = windowStepSizeInDays * DAY_IN_MS;

		Text outKey = new Text();
		Text outVal = new Text();

		SlidingWindow slidingWindow = new SlidingWindow(windowSizeInMS,
				windowStepSizeInMS, DAY_IN_MS);

		Iterator<TimestampPriceDataPoint> valueIterator = values.iterator();
		while (valueIterator.hasNext()) {

			while (!slidingWindow.WindowIsFull() && valueIterator.hasNext()) {

				nextPoint = valueIterator.next();

				TimestampPriceDataPoint pointCopy = new TimestampPriceDataPoint();
				pointCopy.copy(nextPoint);

				try {
					slidingWindow.AddPoint(pointCopy);
				} catch (Exception e) {
					e.printStackTrace();
				}

			}

			if (slidingWindow.WindowIsFull()) {


				LinkedList<TimestampPriceDataPoint> window = slidingWindow
				.GetCurrentWindow();


				double sumX = 0;
				double sumY = 0;
				double sumX2 = 0;
				double sumXY = 0;
				int x = 0;
				for(TimestampPriceDataPoint dataPoint: window)
				{
					x++;
					double y = dataPoint.getClosePrice();
					sumX += x;
					sumY += y;
					sumX2 += x*x;
					sumXY += x*y;
				}
				//Slope(b) = (NΣXY - (ΣX)(ΣY)) / (NΣX2 - (ΣX)2)
				//Intercept(a) = (ΣY - b(ΣX)) / N 
				int n = x;
				double slope = (n*sumXY - sumX*sumY)/(n*sumX2 - sumX*sumX);
				if (Double.isNaN(slope)) slope = 0;
				double intercept =  (sumY -slope*sumX)/n;
				if (Double.isNaN(intercept)) intercept = 0;
				
				String strBackDate = window.getLast().getDate();

				/*
				outKey.set("Code: " + key.getCode() + ", Date: "
						+ strBackDate);
				outVal.set("Equation: " + equation);
				 */
				
				outKey.set(key.getCode() + "\t" +strBackDate);
				outVal.set(slope + "\t" + intercept);
				context.write(outKey, outVal);

				//step window forward

				slidingWindow.slideWindowForward();

			}

		} // end while

	} // end reduce

}
